package io.task.cc.network;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.nio.channels.CompletionHandler;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


/**
 * @author huihui
 */
@Slf4j
@Setter
@Getter
public class WriteCompletionHandler implements CompletionHandler<Integer, WriteCompletionVo> {

    private ChannelContext channelContext = null;
    public final ReentrantLock lock = new ReentrantLock();
    public final Condition condition = lock.newCondition();

    public WriteCompletionHandler(ChannelContext channelContext) {
        this.channelContext = channelContext;
    }

    @Override
    public void completed(Integer bytesWritten, WriteCompletionVo writeCompletionVo) {
        if (bytesWritten > 0) {
            channelContext.channelStat.latestTimeOfSentByte = System.currentTimeMillis();
        }
        if (writeCompletionVo.byteBuffer.hasRemaining()) {
            if (log.isInfoEnabled()) {
                log.info("{} {}/{} has sent", channelContext, writeCompletionVo.byteBuffer.position(), writeCompletionVo.byteBuffer.limit());
            }
            channelContext.asynchronousSocketChannel.write(writeCompletionVo.byteBuffer, writeCompletionVo, this);
        } else {
            handle(bytesWritten, null, writeCompletionVo);
        }
    }

    @Override
    public void failed(Throwable throwable, WriteCompletionVo writeCompletionVo) {
        handle(0, throwable, writeCompletionVo);
    }


    public void handle(Integer bytesWritten, Throwable throwable, WriteCompletionVo writeCompletionVo) {
        ReentrantLock lock = channelContext.writeCompletionHandler.lock;
        lock.lock();
        try {
            channelContext.sendQueueRunnable.canSendNewData = true;
            channelContext.writeCompletionHandler.condition.signal();
            channelContext.channelStat.latestTimeOfSentPacket = System.currentTimeMillis();
            try {
                boolean isSentSuccess = bytesWritten > 0;
                Object attachment = writeCompletionVo.object;
                boolean isPacket = attachment instanceof Packet;
                if (isPacket) {
                    handleOne((Packet) attachment, isSentSuccess);
                } else {
                    List<Packet> packetList = (List<Packet>) attachment;
                    for (Packet packet : packetList) {
                        handleOne(packet, isSentSuccess);
                    }
                }
                if (!isSentSuccess) {
                    Io.close(channelContext, throwable, "写数据返回:" + bytesWritten, CloseCode.WRITE_COUNT_IS_NEGATIVE);
                }
            } catch (Throwable e) {
                log.error(e.toString(), e);
            }

        } finally {
            lock.unlock();
        }

    }


    public void handleOne(Packet packet, Boolean isSentSuccess) {
        PacketMetaData packetMetaData = packet.getPacketMetaData();
        if (packetMetaData != null) {
            packetMetaData.setSentSuccess(isSentSuccess);
        }
        packetMetaData.countDown();

    }

}
